#include "patronus_public.h"
#include "yb_cache.h"
#include <sys/types.h>
#include <sys/syscall.h> 
#include <signal.h>


#define DEFAULT_THREAD_NUM	3

pid_t gettid()
{
	return syscall(SYS_gettid);
}

static int s_interrupted = 0;
static void s_signal_handler (int signal_value)
{
	    s_interrupted = 1;
}

static void s_catch_signals (void)
{
	    struct sigaction action;
	    action.sa_handler = s_signal_handler;
	    action.sa_flags = 0;
	    sigemptyset (&action.sa_mask);
	    sigaction (SIGINT, &action, NULL);
	    sigaction (SIGTERM, &action, NULL);
}


int initialize();
int handle_frontend	(zloop_t *loop, void *socket, void *arg);
int handle_backend	(zloop_t *loop, void *socket, void *arg);

static void server_worker(void *args, zctx_t *ctx, void *pipe);

int
main(int args, char *argv[])
{

	if (-1 == initialize()) {
		TINYLOG_INFO("initialize failed");
		return EXIT_FAILURE;
	}


	zctx_t *ctx = zctx_new();
//	self_session_t *self = (self_session_t *)zmalloc(sizeof(self_session_t));

	 //  Frontend socket talks to clients over TCP
    void *frontend = zsocket_new (ctx, ZMQ_DEALER);
    zsocket_connect (frontend, "tcp://*:8744");

    //  Backend socket talks to workers over inproc 
    void *backend = zsocket_new (ctx, ZMQ_DEALER);
    zsocket_bind (backend, "inproc://workers.thread");

	
	int thread_nbr;
	for (thread_nbr = 0; thread_nbr < DEFAULT_THREAD_NUM; thread_nbr++) {
		zthread_fork(ctx, server_worker, NULL);
	}

	s_catch_signals();
//	zmq_device (ZMQ_QUEUE, frontend, backend);
	zmq_pollitem_t items [] = {
		{ frontend, 0, ZMQ_POLLIN, 0 },
		{ backend,  0, ZMQ_POLLIN, 0 }
	};

	while (1) {
		zmq_poll(items, 2, -1);
		if (items[0].revents & ZMQ_POLLIN) {
//			fprintf(stderr, "frontend in...\n");
			zmsg_t *msg = zmsg_recv (frontend);
			zmsg_send(&msg, backend);
		} else if (items[1].revents & ZMQ_POLLIN) {
//			fprintf(stderr, "backend in...\n");
			zmsg_t *msg = zmsg_recv (backend);
			zmsg_send(&msg, frontend);
		}
		if (s_interrupted) {
			fprintf(stdout, "interrupt is call\n");
			break;
		}

	}
	//	we never get here but clean up anyhow 
	
    zctx_destroy (&ctx);
//	free(self);

	
//	TINYLOG_INFO("EXIT");
	return EXIT_SUCCESS;


}

static void 
server_worker(void *args, zctx_t *ctx, void *pipe)
{
	void *worker = zsocket_new(ctx, ZMQ_DEALER);
	zsocket_connect(worker, "inproc://workers.thread");

	while (1) {
		zmsg_t *msg = zmsg_recv(worker);

		if (msg) {
			//	handle every message with all kinds of types
			fprintf(stderr, "[%d] worker...\n", gettid());
			zmsg_dump(msg);
		}

		if (s_interrupted) {
			fprintf(stdout, "[%d]worker interrupt...\n", gettid());
			break;
		}

	}

	zsocket_destroy(ctx, worker);
//   zctx_destroy (&ctx);
}

int
initialize()
{
	// init tinylog
	tinylog_layout_t * layout = tinylog_new_layout("ybworker_full",
			tinylog_get_layout_type_by_name("full"), NULL);
	void * appender = tinylog_new_appender("ybworker_stderr", "stream");
	tinylog_reg_layout_to_appender(appender, layout);
	tinylog_stream_appender_set_fp(appender, stderr);
	tinylog_reg_appender(appender);

	FILE * fp = fopen("./runtime-ybworker.log", "a");
	if (NULL != fp) {
		layout = tinylog_new_layout("ybworker_full",
				tinylog_get_layout_type_by_name("full"), NULL);
		void * appender = tinylog_new_appender("ybworker_file", "stream");
		tinylog_reg_layout_to_appender(appender, layout);
		tinylog_stream_appender_set_fp(appender, fp);
		tinylog_reg_appender(appender);
	} else {
		TINYLOG_CRIT("can't open log file: %s", strerror(errno));
		return -1;
	}

	//	init yb cache
	if (-1 == ybc_init()) {
		TINYLOG_CRIT("init yb cache failed");
		return -1;
	}

	return 0;

}
